package title2


import java.util.Properties

import org.apache.spark.sql.{DataFrame, Dataset, SaveMode, SparkSession}
import title1.IPUtils

/**
  * 问题2.计算每个省城市成交量的top3（结果保存到MySQL中）
  * SparkSql做法
  */
object Top3ofProvincialCitiesSQL {
  def main(args: Array[String]): Unit = {

    val session = SparkSession.builder().master("local[*]").appName(this.getClass.getName).getOrCreate()
    import session.implicits._

    val ipData: Dataset[String] = session.read.textFile("data/ip.txt")
    val map: Dataset[(Long, Long, String, String)] = ipData.map({
      t =>
        val lines = t.split("[|]")
        val start = lines(2).toLong
        val end = lines(3).toLong
        val province = lines(6)
        val city = lines(7)
        (start, end, province, city)
    })

    // collect  之后，就成了本地的数组
    val ip: Array[(Long, Long, String, String)] = map.collect()

    // 把ip规则的数据集广播出去
    val ipRuleBC = session.sparkContext.broadcast(ip)


    val orderData = session.read.textFile("data/订单数据.log")
    val tempdata = orderData.map({ t =>
      val line: Array[String] = t.split(" ")
      //2.1获取ip
      val ip = line(1)
      val ipnum: Long = IPUtils.ip2Long(ip)
      //2.2获取消费额
      val amount = line(line.length - 1).toInt
      //2.3从广播变量中获取匹配规则
      val ipRulesFromBC: Array[(Long, Long, String, String)] = ipRuleBC.value
      //2.4根据规则获取ip对应省
      val index: Int = IPUtils.searchCity(ipnum, ipRulesFromBC)
      if (index != -1) {
        val province = ipRulesFromBC(index)._3
        val city = ipRulesFromBC(index)._4
        (province, city, amount)
      } else {
        ("unknown", "unknown", 0)
      }
    }).toDF("province", "city", "amount")

    tempdata.createTempView("temptable")
    //统计每个省的城市的成交量
    val tempdata1 = tempdata.sqlContext.sql("select province,city,sum(amount) amount from  temptable group by province,city")
    //窗口函数处理每个省城市  取前3
    tempdata1.createTempView("temptable1")
    val frame = tempdata1.sqlContext.sql("select *,row_number() over(partition by province,city order by amount desc) as rank from temptable1")
    val result = frame.select("province", "city", "amount").where($"rank" <= 3)
    //结果存储到mysql
    val url = "jdbc:mysql://localhost:3306/scott?characterEncoding=utf-8"
    val tname = "Top3ofProvincialCities"
    val p = new Properties()
    p.setProperty("user", "root")
    p.setProperty("password", "123456")
    p.setProperty("driver", "com.mysql.jdbc.Driver")
    result.write.mode(SaveMode.Append).jdbc(url, tname, p)


  }
}
